package com.suxg.hbase.utils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.beanutils.BeanMap;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.KeyOnlyFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.springframework.util.ObjectUtils;

import com.suxg.hbase.api.RowMapper;
import com.suxg.hbase.model.StudentModel;

/**
 * Hbase 工具转换
 */
public class HBaseUtilsMapper {

	private static Connection connection;

	static {
		// localhost
		Configuration configuration = HBaseConfiguration.create();
		configuration.set("hbase.zookeeper.property.clientPort", "2181");
		// 如果是集群 则主机名用逗号分隔
		configuration.set("hbase.zookeeper.quorum", "localhost:2181");
		configuration.set("hbase.nodeParent", "localhost:2181");
		// aliyun
//		Configuration configuration = HBaseConfiguration.create();
//		configuration.set("hbase.zookeeper.property.clientPort", "2181");
//		// 如果是集群 则主机名用逗号分隔
//		configuration.set("hbase.zookeeper.quorum", "101.200.212.148:2181");
//		configuration.set("hbase.nodeParent", "101.200.212.148:2181");

		try {
			connection = ConnectionFactory.createConnection(configuration);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 创建 HBase 表
	 * 
	 * @param tableName      表名
	 * @param columnFamilies 列族的数组
	 */
	public static boolean createTable(String tableName, List<String> columnFamilies) {
		try {
			HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
			if (admin.tableExists(tableName)) {
				return false;
			}
			HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
			columnFamilies.forEach(columnFamily -> {
				HColumnDescriptor columnDescriptor = new HColumnDescriptor(columnFamily);
				columnDescriptor.setMaxVersions(1);
				tableDescriptor.addFamily(columnDescriptor);
			});
			admin.createTable(tableDescriptor);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;
	}

	/**
	 * 删除 hBase 表
	 *
	 * @param tableName 表名
	 */
	public static boolean deleteTable(String tableName) {
		try {
			HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
			// 删除表前需要先禁用表
			admin.disableTable(tableName);
			admin.deleteTable(tableName);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return true;
	}

	/**
	 * 插入数据
	 *
	 * @param tableName        表名
	 * @param rowKey           唯一标识
	 * @param columnFamilyName 列族名
	 * @param qualifier        列标识
	 * @param value            数据
	 */
	public static boolean putRow(String tableName, String rowKey, String columnFamilyName, String qualifier,
			String value) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Put put = new Put(Bytes.toBytes(rowKey));
			put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(qualifier), Bytes.toBytes(value));
			table.put(put);
			table.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;
	}

	/**
	 * 根据 rowKey 获取指定行的数据
	 *
	 * @param tableName 表名
	 * @param rowKey    唯一标识
	 */
	public static <T> T getRow(String tableName, String rowKey, final RowMapper<T> mapper) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Get get = new Get(Bytes.toBytes(rowKey));
			Result result = table.get(get);
			return mapper.mapRow(result, 0, null);
		} catch (Exception e) {
			e.printStackTrace();
		}

		return null;
	}

	/**
	 * 获取指定行指定列 (cell) 的最新版本的数据
	 *
	 * @param tableName    表名
	 * @param rowKey       唯一标识
	 * @param columnFamily 列族
	 * @param qualifier    列标识
	 */
	public static String getCell(String tableName, String rowKey, String columnFamily, String qualifier) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Get get = new Get(Bytes.toBytes(rowKey));
			if (!get.isCheckExistenceOnly()) {
				get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
				Result result = table.get(get);
				byte[] resultValue = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
				return Bytes.toString(resultValue);
			} else {
				return null;
			}

		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * 按条件查询
	 * 
	 * @param <T>
	 * @param tableName
	 * @param rowKey
	 * @param familyCell
	 * @param pamr
	 * @param o
	 * @return
	 */
	public static <T> Object getHbaseObject(String tableName, String rowKey, String familyCell, String pamr, Object o) {
		Table table = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));
			Get get = new Get(Bytes.toBytes(rowKey));
			Result resultCell = table.get(get);
			o = resultCell.getFamilyMap(Bytes.toBytes(familyCell)).get(pamr);
//	  	   
//			Map<byte[], byte[]> familyMap = resultCell.getFamilyMap(Bytes.toBytes(familyCell)).get(pamr);
//			Map<String, String> maps = new HashMap<>();
//			for (Map.Entry<byte[], byte[]> entry : familyMap.entrySet()) {
//				System.out.println(Bytes.toString(entry.getKey()) + "=====>" + Bytes.toString((entry.getValue())));
//				if (ObjectUtils.isEmpty(entry.getValue())) {
//					maps.put(Bytes.toString(entry.getKey()), "");
//				} else {
//					maps.put(Bytes.toString(entry.getKey()), Bytes.toString(entry.getValue()));
//				}
//			}
//			BeanUtils.populate(o, maps);
		} catch (Exception e) {
//			e.printStackTrace();
		} finally {
			// scanner.close();
//			connection.close();
		}
		return o;
	}

	/**
	 * Hbase get   按rowkey 和列族 查询所有
	 * @param <T>
	 * @param tableName 
	 * @param mapper   无用
	 * @param rowKey
	 * @param family
	 * @param o 需要返回的实体对象
	 * @return
	 */
	public static <T> List<T> getHbaseObjectList(String tableName, final RowMapper<T> mapper, String rowKey,
			String family, Object o) {

		Table table = null;
		List<T> rs = new ArrayList<T>();
//		ResultScanner scanner = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));
//			Scan scan = new Scan();
			// 进行全表扫描加开始，结束行，优化查询效率
//            scan.setStartRow(null);
//            scan.setStopRow(null);

//			Get get = new Get(Bytes.toBytes("rowKey1"));
			Get get = new Get(Bytes.toBytes(rowKey));
			Result resultCell = table.get(get);
			// result.listCells();//可以考虑
//		    Map<byte[], byte[]> familyMap = resultCell.getFamilyMap(Bytes.toBytes("student"));
			Map<byte[], byte[]> familyMap = resultCell.getFamilyMap(Bytes.toBytes(family));
			Map<String, String> maps = new HashMap<>();
			// *************************
			for (Map.Entry<byte[], byte[]> entry : familyMap.entrySet()) {
				System.out.println(Bytes.toString(entry.getKey()) + "=====>" + Bytes.toString((entry.getValue())));
				if (ObjectUtils.isEmpty(entry.getValue())) {
					maps.put(Bytes.toString(entry.getKey()), "");
				} else {
					maps.put(Bytes.toString(entry.getKey()), Bytes.toString(entry.getValue()));
				}

			}
			// ***************

//			StudentModel sm = new StudentModel();
			BeanUtils.populate(o, maps);
			rs.add((T) o);
//		   new  BeanMap(familyMap);
//		   BeanUtils.populate(StudentModel.class, familyMap);

//			int caching = scan.getCaching();
//			// 如果caching未设置(默认是1)，将默认配置成5000
//			if (caching == 1) {
//				scan.setCaching(5000);
//			}
//			scanner = table.getScanner(scan);
//			int rowNum = 0;
//			for (Result result : scanner) {
//				rs.add(mapper.mapRow(result, rowNum++,familyMap));
//			}
		} catch (Exception e) {
//			e.printStackTrace();
		} finally {
			// scanner.close();
//			connection.close();
		}
		return rs;
	}

	/**
	 * 检索全表
	 * 
	 * @param <T>
	 *
	 * @param tableName 表名
	 */
	public static <T> List<T> getScanner(String tableName, final RowMapper<T> mapper) {

		Table table = null;
		List<T> rs = new ArrayList<T>();
		ResultScanner scanner = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			// 进行全表扫描加开始，结束行，优化查询效率
//            scan.setStartRow(null);
//            scan.setStopRow(null);

			Get get = new Get(Bytes.toBytes("rowKey1"));
			Result resultCell = table.get(get);
			// result.listCells();//可以考虑
			Map<byte[], byte[]> familyMap = resultCell.getFamilyMap(Bytes.toBytes("student"));
			for (Map.Entry<byte[], byte[]> entry : familyMap.entrySet()) {
				System.out.println(Bytes.toString(entry.getKey()));
			}

			int caching = scan.getCaching();
			// 如果caching未设置(默认是1)，将默认配置成5000
			if (caching == 1) {
				scan.setCaching(5000);
			}
			scanner = table.getScanner(scan);
			int rowNum = 0;
			for (Result result : scanner) {
//				rs.add(mapper.mapRow(result, rowNum++, null));
				 Iterator<KeyValue> it = result.list().iterator();
				  while (it.hasNext()) {
					    KeyValue kv = it.next();
					   System.out.println("@@@@@@@@key@@@@key="+kv.getFamily()+", value="+kv.getValue());
				  }
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			scanner.close();
		}
		return rs;
	}

	/**
	 * 检索表中指定数据
	 *
	 * @param tableName  表名
	 * @param filterList 过滤器
	 */

	public static ResultScanner getScanner(String tableName, FilterList filterList) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			scan.setFilter(filterList);
			return table.getScanner(scan);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * 检索表中指定数据
	 *
	 * @param tableName   表名
	 * @param startRowKey 起始 RowKey
	 * @param endRowKey   终止 RowKey
	 * @param filterList  过滤器
	 */

	public static ResultScanner getScanner(String tableName, String startRowKey, String endRowKey,
			FilterList filterList) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			scan.setStartRow(Bytes.toBytes(startRowKey));
			scan.setStopRow(Bytes.toBytes(endRowKey));
			scan.setFilter(filterList);
			return table.getScanner(scan);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * 删除指定行记录
	 *
	 * @param tableName 表名
	 * @param rowKey    唯一标识
	 */
	public static boolean deleteRow(String tableName, String rowKey) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Delete delete = new Delete(Bytes.toBytes(rowKey));
			table.delete(delete);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;
	}

	/**
	 * 删除指定行的指定列
	 *
	 * @param tableName  表名
	 * @param rowKey     唯一标识
	 * @param familyName 列族
	 * @param qualifier  列标识
	 */
	public static boolean deleteColumn(String tableName, String rowKey, String familyName, String qualifier) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Delete delete = new Delete(Bytes.toBytes(rowKey));
			delete.addColumn(Bytes.toBytes(familyName), Bytes.toBytes(qualifier));
			table.delete(delete);
			table.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;
	}

	/**
	 * 插入数据
	 *
	 * @param tableName        表名
	 * @param rowKey           唯一标识
	 * @param columnFamilyName 列族名
	 * @param pairList         列标识和值的集合
	 */
	public static boolean putRow(String tableName, String rowKey, String columnFamilyName,
			List<Pair<String, String>> pairList) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Put put = new Put(Bytes.toBytes(rowKey));
//            pairList.forEach(pair -> put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(pair.getKey()), Bytes.toBytes(pair.getValue())));

			for (Pair<String, String> pair : pairList) {
				put.addColumn(Bytes.toBytes(columnFamilyName), pair.getFirst().getBytes(), pair.getSecond().getBytes());
			}
			table.put(put);
			table.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;

	}

}
